package com.bee.shepherd.service.etl;

import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.bee.shepherd.entity.WechatGroupChatMsg;
import com.qunar.finance.nbdata.adapter.api.been.ResponseContent;
import com.qunar.finance.nbdata.adapter.api.enums.ServiceStatusCode;
import org.apache.commons.beanutils.BeanUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.lang.reflect.InvocationTargetException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
public class ETLServiceImpl implements IETLService, ApplicationRunner {

    private static final Logger logger = LoggerFactory.getLogger(ETLServiceImpl.class);

    @Resource
    private KafkaCanalConnector kafkaCanalConnector;

    @Resource
    private IESService esService;

    private boolean running;

    private Thread taskThread;

    private static final String INSERT_OP = "INSERT";
    private static final String UPDATE_OP = "UPDATE";
    private static final String DELETE_OP = "DELETE";

    @Override
    public void start() {
        Assert.notNull(kafkaCanalConnector, "connector is null");
        taskThread = new Thread(() -> {
            process();
        }, "KafKaCanalConnectorThread");
        taskThread.setUncaughtExceptionHandler((t, e) -> logger.error("工作线程[ {} ]发生异常:{}", t, e));
        taskThread.start();
        running = true;
    }

    @Override
    public void stop() {
        if (!running) {
            return;
        }
        running = false;
        if (taskThread != null) {
            try {
                taskThread.join();
            } catch (InterruptedException e) {
                // ignore
                logger.warn("interrupt process:{}", e.getMessage());
            }
        }
    }

    @Override
    public void process() {
        while (!running) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                logger.warn("interrupt process:{}", e.getMessage());
            }
        }

        while (running) {
            try {
                kafkaCanalConnector.connect();
                kafkaCanalConnector.subscribe();
                while (running) {
                    try {
                        // 获取message
                        List<FlatMessage> messages = kafkaCanalConnector.getFlatList(100L, TimeUnit.MILLISECONDS);
                        if (messages == null) {
                            continue;
                        }
                        for (FlatMessage message : messages) {
                            long batchId = message.getId();
                            int size = message.getData().size();
                            if (batchId == -1 || size == 0) {
                                continue;
                            } else {
                                // printSummary(message, batchId, size);
                                // printEntry(message.getEntries());
                                ResponseContent content = esOp(message);
                                int code = content.getServiceStatusCode();

                                if(ServiceStatusCode.ERROR == code){
                                    kafkaCanalConnector.rollback();
                                    continue;
                                }
                            }
                        }

                        kafkaCanalConnector.ack(); // 提交确认
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }

        kafkaCanalConnector.unsubscribe();
        kafkaCanalConnector.disconnect();
    }

    private ResponseContent esOp(FlatMessage message) throws InvocationTargetException, IllegalAccessException {
        WechatGroupChatMsg msg = new WechatGroupChatMsg();

        String type = message.getType();
        Map<String,String> data = message.getData().get(0);
        BeanUtils.populate(msg,data);
        logger.info("op type:{} - op data map: {}", type,data);
        switch (type){
            case INSERT_OP:
                esService.save(message);
                break;
            case UPDATE_OP:

                esService.update(message);
                break;
            case DELETE_OP:
                esService.delete(message);
                break;
        }
        return null;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        logger.info("启动etl服务...");

        this.start();
    }
}